/*
*   Coder : Dzlua
*   Email : 505544956@qq.com
*   Time  : 2017/08/08
*/
#pragma once

#include <evpp/tcp_client.h>
#include <evpp/event_loop_thread_pool.h>
#include <evpp/buffer.h>
#include <evpp/tcp_conn.h>

#ifdef WIN32
#include "winmain-inl.h"
#endif

#include "datapkg.hpp"

class Client;

class Session {
public:
    Session(evpp::EventLoop* loop,
            const std::string& serverAddr/*ip:port*/,
            const std::string& name,
            Client* owner)
        : client_(loop, serverAddr, name),
          owner_(owner) {

        client_.SetConnectionCallback(
            std::bind(&Session::OnConnection, this, std::placeholders::_1));
        client_.SetMessageCallback(
            std::bind(&Session::OnMessage, this, std::placeholders::_1, std::placeholders::_2));
    }

    void Start() {
        client_.Connect();
    }

    void Stop() {
        client_.Disconnect();
    }

private:
    void OnConnection(const evpp::TCPConnPtr& conn);

    void OnMessage(const evpp::TCPConnPtr& conn, evpp::Buffer* buf);

private:
    void Send(const evpp::TCPConnPtr& conn, std::shared_ptr<DataPkg> datapkg) {
        if (!datapkg) {
            this->Stop();
            return;
        }
        //
        conn->Send(datapkg->data(), datapkg->size());
        LOG_WARN << "\n-----------------\n";
        LOG_WARN << datapkg->cstring() << "\n";
    }
private:
    evpp::TCPClient client_;
    Client* owner_;
};

class Client {
public:
    Client(const std::string& serverAddr, // ip:port
           int sessionCount,
           int timeout_sec,
           int threadCount,
           std::function<std::shared_ptr<DataPkg>()> get_data)
        : loop_(nullptr),
        session_count_(sessionCount),
        timeout_(timeout_sec),
        connected_count_(0),
        get_data_(get_data) {
        
        loop_ = std::make_shared<evpp::EventLoop>();

        tpool_.reset(new evpp::EventLoopThreadPool(loop_.get(), threadCount));
        tpool_->Start(true);
        
        for (int i = 0; i < sessionCount; ++i) {
            char buf[32];
            snprintf(buf, sizeof buf, "C%05d", i);
            Session* session = new Session(tpool_->GetNextLoop(), serverAddr, buf, this);
            session->Start();
            sessions_.push_back(session);
        }

        if (timeout_sec > 0) {
            loop_->RunAfter(evpp::Duration(double(timeout_sec)), std::bind(&Client::HandleTimeout, this));
        } else {
            loop_->Run();
        }

    }

    ~Client() {
    }

    void OnConnect() {
        if (++connected_count_ == session_count_) {
            LOG_WARN << "all connected";
        }
    }

    void OnDisconnect(const evpp::TCPConnPtr& conn) {
        if (--connected_count_ == 0) {
            LOG_WARN << "all disconnected";

            loop_->QueueInLoop(std::bind(&Client::Quit, this));
        }
    }

    void Run() {
        loop_->Run();
    }

    std::shared_ptr<DataPkg> GetDataPkg() {
        return get_data_();
    }

private:
    void Quit() {
        tpool_->Stop();
        loop_->Stop();
        for (auto &it : sessions_) {
            delete it;
        }
        sessions_.clear();
        while (!tpool_->IsStopped() || !loop_->IsStopped()) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
        tpool_.reset();
    }

    void HandleTimeout() {
        LOG_WARN << "stop";
        for (auto &it : sessions_) {
            it->Stop();
        }
    }
private:
    std::shared_ptr<evpp::EventLoop> loop_;
    std::shared_ptr<evpp::EventLoopThreadPool> tpool_;
    int session_count_;
    int timeout_;
    std::vector<Session*> sessions_;
    std::atomic<int> connected_count_;
    std::function<std::shared_ptr<DataPkg>()> get_data_;
};

void Session::OnConnection(const evpp::TCPConnPtr& conn) {
    if (conn->IsConnected()) {
        conn->SetTCPNoDelay(true);

        this->Send(conn, owner_->GetDataPkg());

        owner_->OnConnect();
    } else {
        owner_->OnDisconnect(conn);
    }
}

void Session::OnMessage(const evpp::TCPConnPtr& conn, evpp::Buffer* buf) {
    this->Send(conn, owner_->GetDataPkg());
}